/***************************************************************************************************
 *Copyright(C),2010-2016,Sumscope
 *FileName	:  DictUpdateManager.cpp
 *Author	:  scofined.qi
 *Version	:  1.0
 *Date		:  2017/04/07
 *Desc		:  //用于主要说明此程序文件完成的主要功能
 *Relation :
 *Others	:  //其他内容说明
 *Function :  //主要函数列表，每条记录应包含函数名及功能简要说明
 *History	:
 * //修改历史记录列表，每条修改记录应包含修改日期、修改者及修改内容简介
 ***************************************************************************************************/
#include "DictUpdateManager.h"

#include <bondlib/BondContainer.h>
#include <core/strings/Rand.h>
#include <core/strings/pinyin_search.h>
#include <core/time/time.h>
#include <qbkernel/include/qb_base.h>
#include <qbprotocol/include/SSEncrypt.h>
#include <qbprotocol/include/SSModel.h>
#include <qbprotocol/include/SSMsgStruct.h>
#include <qbprotocol/include/rdMessage.h>
#include <uam/ServerTimeMgr.h>

#include <QThread>
#include <list>
#include <map>
#include <thread>

#include "DictChecker.h"
#include "DictLoadListener.h"
#include "DictTable.h"
#include "DictThreadGuard.h"

#ifndef WIN32
#include <arpa/inet.h>
#else
#pragma comment(lib, "ws2_32.lib")
#endif
#include "core/sharedata/login_data.h"
#include "qbtools/message/MessagePoster.h"

typedef qb::base::SpellSearch SpellSearch;

#ifdef ENABLE_TIME_METRIC
#define TIME_METRIC_DECLARE() ss::base::TimeRangeMetric _tm_metric;
#define TIME_METRIC_SET_METRICS(STEPS) \
  _tm_metric.SetMetrics(STEPS, _countf(STEPS));
#define TIME_METRIC_ADD_TIME(INDEX) _tm_metric.AddTime(INDEX);
#else
#define TIME_METRIC_DECLARE()
#define TIME_METRIC_SET_METRICS(STEPS)
#define TIME_METRIC_ADD_TIME(INDEX)
#endif

#ifdef WIN32
#else
#define WM_USER 0x0400
#endif

#ifndef MEM_METRIC
#define MEM_METRIC  //
#endif
#ifndef BVIEW
#define BVIEW  //
#endif
#ifndef BS_CFGMETRICS
#define BS_CFGMETRICS  //
#endif
#ifndef BS_METRICS
#define BS_METRICS  //
#endif

#ifndef THREAD_NAME
#define THREAD_NAME  //
#endif

#define BOOLTEST(kProfileDictSimulateModifyPack) true

enum {
  eFT_PL_LOGIN_OK = WM_USER + 5000,  // 6024 登陆系统服务器成功

  eFT_PL_SYS_TIME,  // 6025 系统时间--

  eFT_PL_MARKET_STATUS_OK,   // 6026 市场状态
  eFT_PL_MARKET_STATUS_NOK,  // 6027 市场状态

  eFT_PL_FORCE_QUIT,  // 6028 异地登陆下线--

  eFT_PL_RELOGIN,  // 6029 需要重新登录--

  eFT_PL_HEART_BREAK_SYS_OK,    // 6030 系统服务器连接心跳--
  eFT_PL_HEART_BREAK_SYS_NOK,   // 6031 系统服务器连接心跳--
  eFT_PL_HEART_RECONN_SYS_ING,  // 6032 重新连接系统服务器--
  eFT_PL_FILE_UPDATE,           // 6033 本地列表文件有更新
  eFT_PL_MICRO_UPDATE,          // 6034 宏观数据文件有更新
  eFT_PL_FINANCIALCOMP_UPDATE,  // 6035 机构表有更新
  eFT_PL_HEART_BREAK_SYS_OK2,   // 6036 重联成功标志（心跳灯专用）
  eFT_PL_HEART_BREAK_SYS_RECONN,       // 6037 系统断线，请求重新连接
  eFT_PL_HEART_BREAK_SYS_DELAY,        // 6038 网络延迟
  EFT_PL_HEART_BREAK_SYS_UNOBSTRUCTED  // 6039 网络通畅
};

namespace {
#ifdef WIN32
unsigned __stdcall DictThreadProc(void* lpParameter)
#else
void* DictThreadProc(void* lpParameter)
#endif
{
  // 异常,卡死,杀死
#ifdef WIN32
  __try
#endif
  {
    DictUpdateManager* dict = (DictUpdateManager*)lpParameter;
    dict->DictThreadRun();
  }
#ifdef WIN32
  __except (1)

  {
    // DictThreadGuard::Get().OnException();
    // return 0;
  }
#endif
  log_info("[DICT]字典线程结束");
  DictThreadGuard::Get().OnEnd();
  return 0;
}

}  // namespace
namespace qb {
namespace ht {

//////////////////////////////////////////////////////////////////////////
/*
        PacketAllocator
   中分配的内存,只用于存放收取到的字典更新包数据,属于临时内存
        大量临时内存与全局永久内存混用,会导致内存堆短时间无法释放,也会导致内存碎片
*/
#ifdef USE_HEAP_ALLOCTOR
#define ALLOC_PKT_BUF(SIZE) (char*)m_heap.Alloc(SIZE)
#define FREE_PKT_BUF(BUF, SIZE) m_heap.Free(BUF, SIZE);
#else
#define ALLOC_PKT_BUF(SIZE) new (std::nothrow) char[SIZE];
#define FREE_PKT_BUF(BUF, SIZE) \
  if (BUF) {                    \
    delete[] BUF;               \
  }
#endif
PacketAllocator::PacketAllocator() : m_heap("PacketAllocator") {}
UpdatePacket* PacketAllocator::Create(const UpdatePacket* zipPacket) {
  typedef qb::base::HeapAllocator HeapAllocator;
  class ScopedCharPtr {
    HeapAllocator& m_heap;
    char* m_buffer;
    int m_size;

   public:
    ScopedCharPtr(HeapAllocator& heap, int size) : m_heap(heap), m_size(size) {
      m_buffer = ALLOC_PKT_BUF(size);
      if (m_buffer) memset(m_buffer, 0, size);
    }
    ~ScopedCharPtr() {
      if (m_buffer) {
        FREE_PKT_BUF(m_buffer, m_size);
      }
    }
    char* Pick() {
      char* ptr = m_buffer;
      m_buffer = NULL;
      return ptr;
    }
    operator bool() const { return m_buffer != NULL; }
    operator char*() { return m_buffer; }
  };
  /*
  将zipPacket解压,zipPacket对象只读,之后会马上销毁
  */
  enum kPacketCreate {
    kPacketCreateSuccess,
    kPacketCreateCheckFailed,
    kPacketCreateAllocFailed,
    kPacketCreateUnzipFailed,
  };
  kDictType type = GetDictTypeByVersion(zipPacket->szPatchName);
  if (type >= kDictTypeCount) {
    log_error(
        "[DICT][ERROR]invalid kDictType:%d seq:%d updateid:%d patch:%s md5:%s "
        "len:%d origsize:%d",
        type, zipPacket->sequence, zipPacket->updateId, zipPacket->szPatchName,
        zipPacket->szMd5, zipPacket->fileLen, zipPacket->origsize);
    return NULL;
  }

  if (!qb::CheckMd5(zipPacket->fileBuf, zipPacket->fileLen, zipPacket->szMd5)) {
    log_error(
        "[DICT][ERROR]Failed  status:%d seq:%d updateid:%d patch:%s md5:%s "
        "len:%d origsize:%d",
        kPacketCreateCheckFailed, zipPacket->sequence, zipPacket->updateId,
        zipPacket->szPatchName, zipPacket->szMd5, zipPacket->fileLen,
        zipPacket->origsize);
    return NULL;
  }

  int nTotalSize = zipPacket->origsize + 32;
  kPacketCreate status = kPacketCreateAllocFailed;
  ScopedCharPtr buffer(m_heap, nTotalSize);
  if (buffer) {
    int oSize = 0;
    if (rdMessage::UnCompress((char*)zipPacket->fileBuf, zipPacket->fileLen,
                              buffer, &oSize, nTotalSize)) {
      if (oSize != zipPacket->origsize) {
        log_error(
            "[DICT][ERROR]UnCompress success,oSize:%d,seq:%d updateid:%d "
            "patch:%s md5:%s len:%d origsize:%d",
            oSize, zipPacket->sequence, zipPacket->updateId,
            zipPacket->szPatchName, zipPacket->szMd5, zipPacket->fileLen,
            zipPacket->origsize);
      }
      UpdatePacket* rawPacket = new (std::nothrow) UpdatePacket();
      if (rawPacket) {
        *rawPacket = *zipPacket;
        rawPacket->fileBuf = buffer.Pick();
        rawPacket->fileLen = oSize;
        rawPacket->finish = false;
        rawPacket->type = type;
        rawPacket->origsize = zipPacket->fileLen;
        if (rawPacket->fileBuf != nullptr) {
          status = kPacketCreateSuccess;
          return rawPacket;
        }
        delete rawPacket;
        status = kPacketCreateAllocFailed;
      }
    } else
      status = kPacketCreateUnzipFailed;
  }
  if (status != kPacketCreateSuccess) {
    log_error(
        "[DICT][ERROR]Failed  status:%d seq:%d updateid:%d patch:%s md5:%s "
        "len:%d origsize:%d",
        status, zipPacket->sequence, zipPacket->updateId,
        zipPacket->szPatchName, zipPacket->szMd5, zipPacket->fileLen,
        zipPacket->origsize);
  }

  return NULL;
}
UpdatePacket* PacketAllocator::CreateFinishPacket(
    const UpdatePacket* zipPacket) {
  UpdatePacket* rawPacket = new (std::nothrow) UpdatePacket();
  if (rawPacket) {
    //*rawPacket = *zipPacket;
    memset(rawPacket, 0, sizeof(UpdatePacket));
    rawPacket->sequence = zipPacket->sequence;
    rawPacket->updateId = zipPacket->updateId;
    FIELDCOPY(rawPacket->szCategory, zipPacket->szCategory);
    FIELDCOPY(rawPacket->szPatchName, zipPacket->szPatchName);
    FIELDCOPY(rawPacket->szFileName, zipPacket->szFileName);
    rawPacket->origsize = zipPacket->origsize;
    FIELDCOPY(rawPacket->szMd5, zipPacket->szMd5);
    rawPacket->fileLen = zipPacket->fileLen;
    rawPacket->fileBuf = NULL;

    rawPacket->finish = true;
    rawPacket->type = GetDictTypeByVersion(zipPacket->szPatchName);
  }
  return rawPacket;
}
void PacketAllocator::Release(UpdatePacket* rawPacket) {
  if (rawPacket) {
    if (rawPacket->fileBuf) {
      FREE_PKT_BUF(rawPacket->fileBuf, rawPacket->fileLen);
    }
    delete rawPacket;
  }
}
void PacketAllocator::Reset() { m_heap.Reset(); }

class DictReqProgress {
  struct DictReqInfo {
    DWORD start;  // 开始请求时间
    DWORD end;    // 最大请求时间
    int count;    // 当前已经收取的消息个数
    bool skip;    // true,表示有忽略消息
  };
  kDictType m_progress;   //
  int m_skip_count;       //
  bool m_emergent_check;  //
  DWORD m_max_rsp_time;   //
  DictReqInfo m_dicts[kDictTypeCount];

 public:
  DictReqProgress(DWORD maxRspTime) : m_max_rsp_time(maxRspTime) { Reset(); }
  void Reset() {
    m_progress = kDictType(0);
    m_skip_count = 0;
    m_emergent_check = false;
    for (int i = 0; i < kDictTypeCount; i++) {
      DictReqInfo& di = m_dicts[i];
      di.start = di.end = 0;
      di.count = 0;
      di.skip = false;
    }
  }
  void AddProgress(kDictType type) {
    BASSERT(type >= 0 && type < kDictTypeCount);
    m_progress = type;
    m_dicts[type].start = qb::base::GetHighResTime();
    m_dicts[type].end = m_max_rsp_time + m_dicts[type].start;
  }
  int GetSkipCount() const { return m_skip_count; }
  void AddSkipDict(kDictType type) {
    BASSERT(type >= 0 && type < kDictTypeCount);
    if (!m_dicts[type].skip) {
      m_dicts[type].skip = true;
      m_skip_count++;
      m_emergent_check = true;
    }
  }
  void SetMsgCount(kDictType type, int count) {
    BASSERT(type >= 0 && type < kDictTypeCount);
    m_dicts[type].count = count;
  }
  int GetMsgCount(kDictType type) const {
    BASSERT(type >= 0 && type < kDictTypeCount);
    return m_dicts[type].count;
  }
  DWORD GetRspTime(kDictType type) {
    if (m_dicts[type].start != 0) {
      return qb::base::GetHighResTime() - m_dicts[type].start;
    }
    return 0;
  }
  bool IsNeedEmergentCheck() const {
    return m_skip_count > 0 && m_emergent_check;
  }
  void SetEmergentCheck(bool check) { m_emergent_check = check; }
};
class UpdatePackHistory {
  typedef std::list<UpdatePacket> PacketInfoList;
  typedef std::map<kDictType, PacketInfoList> PacketInfoListMap;
  PacketInfoListMap m_packet_infos;  // 只有头部信息,没有fileBuf实质内容
  int m_prefix_count;

 public:
  UpdatePackHistory() : m_prefix_count(1) {}
  void Add(const UpdatePacket* packet) {
    if (packet->finish) return;
    PacketInfoListMap::iterator it = m_packet_infos.find(packet->type);
    if (it == m_packet_infos.end()) {
      m_packet_infos[packet->type] = PacketInfoList();
      it = m_packet_infos.find(packet->type);
    }
    PacketInfoList& pilist = it->second;

    UpdatePacket info = *packet;
    info.fileBuf = NULL;
    pilist.push_back(info);
  }
  bool CheckState(kDictType type, DictRequest& req) {
    // DictRequest 已经按照字典检查结果纠正了
    // 此处,再对纠正结果,按照收取到的字典包信息再次纠正下
    // 确保updateid与md5严格对应
    // 由于合并包的影响,updateid与md5可能并不是严格对应
    // 返回true表示修正过了,false表示不做修正
    PacketInfoListMap::iterator it = m_packet_infos.find(type);
    if (it == m_packet_infos.end()) return false;
    PacketInfoList& pilist = it->second;
    if (pilist.size() <= 0) return false;

    // PacketInfoList 是updateId严格递增排序,如下反向遍历
    bool updated = false;
    int nprefix = m_prefix_count;
    for (PacketInfoList::const_reverse_iterator it = pilist.rbegin();
         it != pilist.rend(); ++it) {
      const UpdatePacket& packet = *it;
      if (req.updateId < packet.updateId) continue;

      req.updateId = packet.updateId;
      FIELDCOPY(req.szMd5, packet.szMd5);
      updated = true;
      nprefix--;
      if (nprefix < 0) break;
    }
    pilist.clear();
    return updated;
  }
};

}  // namespace ht
}  // namespace qb

// #define USE_CONDVARIABLE
#ifdef USE_CONDVARIABLE
#include <chrono>
#include <condition_variable>
#include <mutex>
class SyncPtrQueue {
  std::mutex m_mutex;
  std::condition_variable m_variable;
  std::list<void*> m_pointers;

 public:
  void Add(void* ptr) {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_pointers.push_back(ptr);
    lock.unlock();
    m_variable.notify_all();
  }
  void* Pick(int msTime = 1) {
    void* ptr = 0;
    std::unique_lock<std::mutex> lock(m_mutex);
    if (std::cv_status::timeout ==
        m_variable.wait_for(lock, std::chrono::microseconds(msTime))) {
      if (!m_pointers.empty()) {
        ptr = m_pointers.front();
        m_pointers.pop_front();
      }
    }
    return ptr;
  }
  int Size() {
    std::unique_lock<std::mutex> lock(m_mutex);
    return m_pointers.size();
  }
};
#else
class SyncPtrQueue {
  qb::base::SyncPtrQueue<void> m_queue;
  qb::base::Event m_event;

 public:
#ifdef WIN32
  SyncPtrQueue() : m_event(false, false) {}
#else
  SyncPtrQueue() {}
#endif
  void Add(void* ptr) {
    m_queue.Add(ptr);
    m_event.Signal();
  }
  bool WaitData(int msTime = 20) {
    // 空转等待数据到来,用于减少GetTickCount调用,优化性能
#ifdef WIN32
    return WAIT_OBJECT_0 == m_event.Wait(msTime);
#else
    return 0 == m_event.Wait(msTime);
#endif
  }
  void* Pick(int msTime = 20) {
    void* data = m_queue.Pick();
    if (data == nullptr) {
#ifdef WIN32
      if (WAIT_OBJECT_0 == m_event.Wait(msTime))
#else
      if (0 == m_event.Wait(msTime))
#endif
      {
        data = m_queue.Pick();
      }
    }
    return data;
  }
  int Size() { return m_queue.Size(); }
};
#endif

namespace {
const char* GetStateName(DictUpdateManager::kState state) {
  static const char* names[DictUpdateManager::kStateCount] = {"IDLE", "REQ",
                                                              "CHK", "EXIT"};
  return names[state];
}
const char* GetPullTypeName(DictUpdateManager::kPullType type) {
  static const char* names[DictUpdateManager::kPullTypeCount + 1] = {
      "首次请求", "收到信号", "检查失败", "断网重连", "非法状态"};
  if (type < 0 || type >= DictUpdateManager::kPullTypeCount)
    return names[DictUpdateManager::kPullTypeCount];
  return names[type];
}
}  // namespace

//////////////////////////////////////////////////////////////////////////

using namespace qb;

IDictUpdate* getDictUpdate() { return &DictUpdateManager::instance(); }

DictUpdateManager::DictUpdateManager()
    : m_signal(1),
      m_eixting(false),
      m_is_requesting(false),
      m_req_count(0),
#ifdef CCORE_LINUX
      m_thread(-1),
#else
      m_thread(NULL),
#endif
      m_unloaded(false),
      m_connect_status(kConnectNormal),
      m_first_wait(false),
      m_checker(NULL),
      m_req_time(qb::base::GetHighResTime()),
      m_current_msgcount(0),
      m_state(kStateIdle),
      // m_raw_queue(AfxGetConfig().GetInt(kProfileLockSpinCount)),
      // m_check_queue(AfxGetConfig().GetInt(kProfileLockSpinCount)),
      m_max_rsp_time(300000),  // 5分钟,
      m_last_conreq_time(0),
      m_state_start(0),
      m_check_latter(0),
      m_index_init(false),
      m_max_check_time(10 * 60 *
                       1000),  // 检查如果超过此时间，将认定为失败,默认10分钟
      m_conreq_gap(15 * 60 * 1000),  // 重连15分钟后字典检查
      m_signal_gap(qb::base::Rand::RandInt(10000, 150000)) {
  m_raw_queue = new SyncPtrQueue();
  m_max_rsp_time = 60000;  // AfxGetConfig().GetInt(kProfileDictMaxRspTime);
  m_req_progress = new DictReqProgress(m_max_rsp_time);
  m_history = NULL;  // new UpdatePackHistory();
  m_event = qb::EventWrapper::Create(true);
  m_event->Reset();
}

DictUpdateManager::~DictUpdateManager() {
  WaitForFinish();
  if (m_req_progress) delete m_req_progress;
  if (m_history) delete m_history;
  m_history = NULL;
  m_checker = NULL;
  m_req_progress = NULL;
  if (m_raw_queue) delete m_raw_queue;
  m_raw_queue = nullptr;
}

void DictUpdateManager::init() {
  qbBase::getMessageKernel()->RegisterMessageKernelObserver(this);
  qbBase::getMessageProcess()->RegisterMsgProcessorOnThread(
      E_FID_CT_CLIENT_FILEUPDATE_MSG_REQ, this);
  qbBase::getMessageProcess()->RegisterMsgProcessorOnThread(
      E_FID_CT_CLIENT_FILEUPDATE_FIN, this);
  qbBase::getNetworkService()->RegisterNetworkEventObserver(this);

  std::thread t([this]() { firstLoad(); });
  t.detach();
}

void DictUpdateManager::uninit() {
  qbBase::getNetworkService()->DeRegisterNetworkEventObserver(this);
  qbBase::getMessageProcess()->UnRegisterMsgProcessorOnThread(
      E_FID_CT_CLIENT_FILEUPDATE_MSG_REQ, this);
  qbBase::getMessageProcess()->UnRegisterMsgProcessorOnThread(
      E_FID_CT_CLIENT_FILEUPDATE_FIN, this);
  qbBase::getMessageKernel()->DeRegisterMessageKernelObserver(this);
}

// void DictUpdateManager::Initialize()
//{
//	/*
//	初始化字典线程和解包线程,然后给服务器发送字典请求
//	解包线程:只用于将收取到的字典更新包进行解压缩,将解压缩后的更新包放在指定队列中
//	字典线程:负责对解压缩后的更新包进行处理,读写本地字典文件
//	*/
//	SendFirstRequest();
// }

namespace {
class LocalDictState {
  struct DictState {
    kDictType type;
    std::string name;
    std::string md5;
    int updateid;
  };
  std::vector<DictState> m_states;

 public:
  LocalDictState() {
    m_states.reserve(kDictTypeCount);
    Add("QBBondListV19", "EA07EC6C39205D004A141B16AB16FA8081599EB8", 547803);
    Add("QBBondPrimaryInfoV4", "4C476E689D065A76D49B8C19DE344B685798CB4C",
        547804);
    Add("QBBondAdditionInfoV3", "AF073FB9E6259B84857101E6C29AC84143BCA85E",
        547543);
    Add("BondDictV20", "712C0AABF760757C3DF31C468DBD4AB5593B8D4C", 544308);
    Add("HolidayV3", "09E812F0B15C08F44391C21C12BE48DD036DF0EF", 547613);
    Add("MacroIndex", "B472E9A68B4208EDA9CCF8A2F79CB0750681551F", 547644);
    Add("QBIssuerInfoV5", "52AADDF6F5D880067C08981A6C7D520087FCA1CA", 547806);
    Add("QBProvinceInfoV3", "DB2FE51585B0CC414E2BD27F780E7C58804029F0", 547807);
    Add("QBSwSectorInfoV2", "8195CA08B0A6331A8BFFBC3F15B1D3E8AE5AE12C", 296265);
    Add("UnderwriterV3", "0F8A075E56A652EB67E94D07CA4765F916FF8947", 469512);
    Add("FinancialCompanyV1", "7C95B2060AAE3280A4ED10622D41E2BC62026EC1",
        547494);
    Add("ContinuousContractInfoV2", "F21419DE003C88D44BDFC91FEED8549BB7CCB445",
        546358);
    Add("ExchangeRepoV2", "81AE5F9EC32DE32CAEC41A911E9059C221C0928C", 296268);
    Add("BondCDCAuth", "E3838770AF80C1A55576E7EDBD5F0E6DD1BCAE86", 547601);
  }
  ~LocalDictState() {}
  void Add(const char* name, const char* md5, int id) {
    DictState state;
    state.type = (kDictType)m_states.size();
    state.name = name;
    state.md5 = md5;
    state.updateid = id;
    m_states.emplace_back(state);
  }
  const char* Get(kDictType type, int& updateid) {
    BASSERT(type >= 0 && type < (int)m_states.size());
    updateid = m_states[type].updateid;
    return m_states[type].md5.c_str();
  }
};

// 模拟本地向服务器发送特定包请求,用于复现场景排查问题
// 正式使用时,替换SendDictRequest中的m_dict_manager.GetDictState(type,
// updateid);
const char* GetCurrentDictState(kDictType type, int& updateid) {
  static LocalDictState states;
  return states.Get(type, updateid);
}
}  // namespace

static void RequestDictUpdate(int id, const char* version, const char* md5) {
  if (!version || !md5) return;

  xFileUpdateReq_t xBody;
  memset(&xBody, 0x00, sizeof(xBody));
  xBody.updateId = id;
  FIELDCOPY(xBody.szMd5, md5);
  FIELDCOPY(xBody.szPatchName, version);
  FIELDCOPY(xBody.szCategory, "DataDictionary");

  xMsgHeadReq_t xhead;
  xhead.msgtype = htonl(E_FID_CT_CLIENT_FILEUPDATE_MSG_REQ);
  xBody.updateId = htonl(xBody.updateId);

  char sndBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN] = {0};
  int off = 0;
  memset(sndBuf, 0x00, Y_RD_SERVICE_MESSAGE_BLOCK_LEN);
  rdMessage::bPack((char*)&xhead, sizeof(xhead), (char*)&xBody, sizeof(xBody),
                   sndBuf, &off);
  qbBase::getMessageKernel()->Send(loginData::instance().ConnectID(), sndBuf,
                                   off);
}

void DictUpdateManager::SendDictRequest(kDictType type) {
  BASSERT(type >= 0 && type < kDictTypeCount);
  SCOPED_LOCK(m_lock);

  char szMd5[64] = {0};
  const DictDesc& dd = DictDesc::GetStockDictDesc(type);
  int updateid = 0;
  const char* md5 = m_dict_manager.GetDictState(type, updateid);
  FIELDCOPY(szMd5, md5);
  auto it = m_check_results.find(type);
  if (it != m_check_results.end()) {
    const QBDictCheck_Ack& ack = it->second.ack;
    if (ack.m_UpdateId < updateid && dd.GetType() != kDictTypeMacroIndex) {
      updateid = ack.m_UpdateId;
      FIELDCOPY(szMd5, ack.m_Md5);

      log_info("[DICT]Request Patch changed %20s  %s-%d ==> %s-%d",
               dd.m_version.c_str(), md5, updateid, ack.m_Md5, ack.m_UpdateId);
      if (m_history) {
        log_info("[DICT]Request Patch changed2 %20s  %s-%d ==> %s-%d",
                 dd.m_version.c_str(), ack.m_Md5, ack.m_UpdateId, szMd5,
                 updateid);
      }
    }
  }

  RequestDictUpdate(updateid, dd.m_version.c_str(), szMd5);
  log_debug("[DICT]Request Patch:%20s md5:%s-%d", dd.m_version.c_str(), szMd5,
            updateid);
  m_current_type = type;
  m_current_msgcount = 0;
  SetState(kStateReq);
  m_req_time = qb::base::GetHighResTime();  // 当前字典的请求时间
  m_req_progress->AddProgress(m_current_type);
}
void DictUpdateManager::SendFirstRequest(kPullType type) {
  m_req_count++;  // 仅供调试用
  m_signal.Reset();
  m_req_progress->Reset();
  log_info("[DICT]基于原因:%s  开始字典更新", GetPullTypeName(type));
  SetState(kStateReq);
  m_dict_manager.ClearSeqSet();
  SendDictRequest(kDictTypeBondList);
}
void DictUpdateManager::OnCheckReceived(void* wParam, void* lParam) {
  SSAckMsg* pAckMsg = (SSAckMsg*)wParam;
  DictCheckAck* ack = new DictCheckAck();
  ack->reqid = pAckMsg->m_RequestID;
  ack->ack = *(QBDictCheck_Ack*)pAckMsg->m_pRet;
  m_check_queue.Add(ack);
  log_trace(
      "[DICT][CHECK]收到检查结果 reqid:%d patch:%s code:%d md5:%s updateid:%d",
      ack->reqid, ack->ack.m_PatchName, ack->ack.m_retcode, ack->ack.m_Md5,
      ack->ack.m_UpdateId);
}
void DictUpdateManager::OnPackReceived(int32_t wParam, void* lParam) {
  /*
  // UI线程,网络回调线程
  从网络中收取到字典请求的响应包,主线程,不是字典线程
  因此调用字典里的函数需要加锁
  */
  bool success = false;
  UpdatePacket* zipPacket = (UpdatePacket*)lParam;
  if (zipPacket != NULL && zipPacket->fileLen > 0) {
    UpdatePacket* rawPacket = m_allocator.Create(zipPacket);
    if (rawPacket) {
      BASSERT(rawPacket->type != kDictTypeCount);
      BASSERT(rawPacket->finish == false);
      BASSERT(rawPacket->fileBuf != NULL);
      m_raw_queue->Add(rawPacket);
      success = true;
      if (m_first_wait) m_progresser.AddTotal(rawPacket->origsize);
    }
  } else if (E_FID_CT_CLIENT_FILEUPDATE_FIN == wParam && zipPacket) {
    UpdatePacket* rawPacket = m_allocator.CreateFinishPacket(zipPacket);
    if (rawPacket) {
      log_warning(
          "[DICT]Response Patch:%20s md5:%s-%d origsize:%6d fileLen:%6d",
          zipPacket->szPatchName, zipPacket->szMd5, zipPacket->updateId,
          zipPacket->origsize, zipPacket->fileLen);
      BASSERT(rawPacket->type != kDictTypeCount);
      BASSERT(rawPacket->finish == true);
      BASSERT(rawPacket->fileBuf == NULL);
      m_raw_queue->Add(rawPacket);
      success = true;
      if (m_first_wait) m_progresser.AddTotal(-1);
    }
  }

  if (!success) {
    BASSERT(0);
    log_error("[DICT][ERROR]invalid update packet:wParam:%d zipPacket:0x%x",
              wParam, zipPacket);
  }
}

void DictUpdateManager::OnSignalReceived() {
  /*
  // UI线程,网络回调线程==>等待字典线程
  从网络中收取到服务器发送的字典更新通知包,需要QB请求最新字典
  //
  如果字典线程正在NotifyFinished中通知UI线程,而UI线程恰好在此时触发当前函数调用
  // 就可能导致死锁
  // 旧代码如下:
  SCOPED_LOCK(m_lock);
  m_signal_count++;
  m_signal_tick = GetTickCount();
  */
  m_signal.Add();
}
void DictUpdateManager::OnConnectSignal(int id) {
  /*
  // UI线程,网络回调线程
  参数:int pFunid = (int)wParam;
  eFT_PL_RELOGIN
  重复登录，强制下线,客户端被断开，提示客户 eFT_PL_FORCE_QUIT
  重复登录，强制下线,客户端被断开，提示客户 eFT_PL_HEART_RECONN_SYS_ING
  正在重连 eFT_PL_HEART_BREAK_SYS_NOK		网络连接断开
  eFT_PL_HEART_BREAK_SYS_OK2		网络重连成功
  eFT_PL_HEART_BREAK_SYS_OK		连接成功
  eFT_PL_HEART_BREAK_SYS_RECONN	系统断线，请求重新连接
  */
  class ConnectMsgNames {
    struct MsgInfo {
      int id;
      std::string name, text;
    };
    std::vector<MsgInfo> m_messages;

   public:
    ConnectMsgNames() {
      Add(eFT_PL_RELOGIN, "eFT_PL_RELOGIN",
          "重复登录，强制下线,客户端被断开，提示客户");
      Add(eFT_PL_FORCE_QUIT, "eFT_PL_FORCE_QUIT",
          "重复登录，强制下线,客户端被断开，提示客户");
      Add(eFT_PL_HEART_RECONN_SYS_ING, "eFT_PL_HEART_RECONN_SYS_ING",
          "正在重连");
      Add(eFT_PL_HEART_BREAK_SYS_NOK, "eFT_PL_HEART_BREAK_SYS_NOK",
          "网络连接断开");
      Add(eFT_PL_HEART_BREAK_SYS_OK2, "eFT_PL_HEART_BREAK_SYS_OK2",
          "网络重连成功");
      Add(eFT_PL_HEART_BREAK_SYS_OK, "eFT_PL_HEART_BREAK_SYS_OK", "连接成功");
      Add(eFT_PL_HEART_BREAK_SYS_RECONN, "eFT_PL_HEART_BREAK_SYS_RECONN",
          "系统断线，请求重新连接");
    }
    int GetIndex(int id) {
      for (int i = 0; i < (int)m_messages.size(); i++)
        if (m_messages[i].id == id) return i;
      return -1;
    }
    const char* GetName(int index) { return m_messages[index].name.c_str(); }
    const char* GetText(int index) { return m_messages[index].text.c_str(); }
    void Add(int id, const char* name, const char* text) {
      MsgInfo mi;
      mi.id = id;
      mi.name = name;
      mi.text = text;
      m_messages.push_back(mi);
    }
  };
  static ConnectMsgNames msgname;
  int index = msgname.GetIndex(id);
  if (index >= 0)
    log_debug("[DICT] %d %s %s", id, msgname.GetName(index),
              msgname.GetText(index));

  if (eFT_PL_HEART_BREAK_SYS_OK == id)  // 重新连接后
  {
    m_connect_status = kConnectRecover;
  } else if (eFT_PL_HEART_BREAK_SYS_NOK == id)  // 断线后
  {
    m_connect_status = kConnectLost;
  } else if (eFT_PL_FORCE_QUIT == id)  // 被踢
  {
    m_connect_status = kConnectLostKicked;
  }
}

void DictUpdateManager::firstLoad() {
  m_dict_manager.Init();
  if (m_dict_manager.HasBadDict()) {
    m_needSyncFromServer = true;
    m_event->Set();
    log_warning("DictUpdateManager::firstLoad, Dict has problem!");
    return;
  }

  NotifyFinished(true);
  BuildSpellIndex();
  log_warning("DictUpdateManager::firstLoad, Dict is OK!");
  m_event->Set();
}

int DictUpdateManager::WaitFirstLoad(int start, int end) {
  // 登录过程被踢,登录过程网络断开,登录过程获取字典更新包超时,都会返回负值
  /*
  notify 是要通知的窗口,通常是登录窗口
  msg:
  start:
  end:

  登录进度通知:
          本地文件检查,获取最新版本信息
          发送数据更新请求
          数据更新请求收取完毕
          数据更新请求刷新
          更新并通知到数据容器

          各个类型的数据包的通常数据大小比例
          当前版本检查
  线程 Login_NetEngineBeginWorking 会等待字典线程来进行初次字典更新
  */
  m_event->Wait(0xFFFFFFFF);
  static int call_count = 0;
  call_count++;

#ifdef CCORE_LINUX
  if (m_thread != -1)
#else
  if (m_thread != NULL || m_first_wait)  // 可能登录过程中超时
#endif
  {
    log_warning("[DICT]WaitFirstLoad已经被调用%d次", call_count);
    return kProgressOk;
  }
  DWORD wait_max = 200000;  // AfxGetConfig().GetInt(kProfileDictFirstWaitTime);
  DWORD wait_start = qb::base::GetHighResTime();
  DWORD wait_end = wait_start + wait_max;
  BS_METRICS("[DICT]WaitFirstLoad");
  m_first_wait = true;
  m_dict_manager.Init();  // 可以多遍
  m_progresser.Init(start, end);
  // 评估工作量,只能有两者状态,全量和非全量
  m_progresser.SetDataSize(m_dict_manager.PredictProgressSize());
  // m_progresser.AddTotal(/*真实需要更新的数据大小从服务器请求而来*/);
  // m_progresser.AddTotal(-1);
#ifdef USE_CONDVARIABLE
  log_info("[DICT]开始拉取字典数据(使用std::mutex+condition_variable)...");
#else
  log_info("[DICT]开始拉取字典数据(使用ss::base::SyncPtrQueue<>)...");
#endif
  SendFirstRequest(kPullTypeOnInit);
  BASSERT(GetState() == kStateReq);
  int packets = 0;      // 总共处理的更新包个数
  int times = 0;        // 总共循环多少批次
  int maxlength = 0;    // 更新包队列的最大长度
  int maxprocess = 0;   // 单批次处理封包最大个数
  int sleep_count = 0;  // 休眠次数
  int check_count = 0;  // 记录GetTickCount次数,过多将导致性能额外开销
  int maxsleepgap = 0;       // 记录最大的休眠间隔时间
  int multi_fail_count = 0;  // 连续失败次数
  const int multi_fail_max = 10000;
  DWORD net_check = qb::base::GetHighResTime();
  DWORD last_sleep = net_check;
  bool net_status = true;
  HighResTime timer;
  DWORD worktime = 0;
  while (1) {
    bool sleeped = false;
    int length = m_raw_queue->Size();
    maxlength = maxlength < length ? length : maxlength;
    int processed = length <= 0 ? 0
                                : ProcessRawPackets(m_needSyncFromServer,
                                                    qb::base::GetHighResTime(),
                                                    200, &worktime);
    multi_fail_count = processed <= 0 ? (multi_fail_count + 1) : 0;
    packets += processed;
    times++;
    length = processed > 0 ? m_raw_queue->Size() : length;
    maxprocess = maxprocess < processed ? processed : maxprocess;
    if (processed <= 0 && GetState() == kStateIdle && length <= 0) {
      // 成功
      break;
    }

    if (processed <= 0 && multi_fail_count > multi_fail_max) {
      // 如下等待,大幅度降低GetTickCount的次数,使得原先600-700ms
      // 降低到100-150ms左右,并降低CPU占用率
      if (!m_raw_queue->WaitData(10)) {
        check_count++;
        DWORD tick = qb::base::GetHighResTime();
        if (maxsleepgap < (int)(tick - last_sleep))
          maxsleepgap = tick - last_sleep;
        if (tick > last_sleep + 50) {
          last_sleep = tick;
          sleep_count++;
          sleeped = true;
          QThread::sleep(0);
        }
      }
    }

    // 每次成功后,或者失败间隔1秒钟检测一次网络状态
    if (sleeped) {
      DWORD tick = last_sleep;
      if (tick > net_check + 1000) {
        net_check = tick;
        if (length <= 0 && CheckConnectionLost()) {
          log_error("[DICT]网络连接断开");
          net_status = false;
          SetState(kStateIdle);
          break;
        }
      }
      if (tick > wait_end) {
        // 超时失败,可以在后续字典校验功能中恢复
        log_warning(
            "[DICT]收取字典更新包超时,已经等待时间:%d ms,预设等待时间:%d ms",
            tick - wait_start, wait_max);
        SetState(kStateIdle);
        break;
      }
    }
  }

  unsigned int total_time = timer.elapsed().microsec();
  double work_ratio =
      total_time <= 0 ? 0 : (((double)worktime * 1000) * 100.0 / total_time);
  log_debug(
      "[DICT]首次登录处理更新包%d个,%d批次 %d检查  %4d休眠 %6d,总共耗时%dus  "
      "实际解码时间:%dus  占比:%.2f 队列最大长度:%d 单次处理最多个数:%d",
      packets, times, check_count, sleep_count, maxsleepgap, total_time,
      worktime * 1000, work_ratio, maxlength, maxprocess);
  // 更新数据到各个容器中,启动后来收取线程
  if (!net_status) {
    log_error("[DICT]收取字典碰到网络断开");
    if (m_connect_status == kConnectLostKicked) return kProgressLoginKicked;
    return kProgressDisconnected;
  }

  {
    HighResTime syntimer;
    MEM_METRIC("NotifyFinished(true)");
    SCOPED_LOCK(m_lock);
    NotifyFinished(m_needSyncFromServer);

    if (GetState() != kStateReq) LoadDictCache(false);
    log_info("[DICT]更新到内存中消耗:%8dus", syntimer.elapsed().microsec());
  }

  if (0)  // 将一些内存打印出来
  {
    CBondContainer::instance().DumpStatus();
    DumpStatus();
  }

  log_info("[DICT]新字典总共耗时 %dms",
           (qb::base::GetHighResTime() - wait_start));
  log_info("[DICT]字段更新网络检查 %s,信号计数:%d",
           net_status ? "OK" : "Failed", m_signal.GetSignal());
  m_progresser.Finish();
  m_first_wait = false;
  StartDictThread();
  return kProgressOk;
}

bool DictUpdateManager::StartDictThread(bool force) {
  // 启动字典后台线程
  SCOPED_LOCK(m_lock);

#ifdef WIN32
  if (!m_thread || force) {
    if (m_thread) {
      CloseHandle(m_thread);
      m_thread = NULL;
    }

    if (force && m_checker) {
      m_checker->SetCheckLatter("字典线程被强制重启");
    }

    unsigned threadID;
    m_thread = (void*)_beginthreadex(NULL, 0, &DictThreadProc, (void*)this, 0,
                                     &threadID);
    log_info("[DICT]开启字典线程");
  }
#else
#ifdef CCORE_LINUX
  if (m_thread == -1 || force) {
    if (m_thread != -1) {
      m_thread = -1;
    }
#else
  if (m_thread == NULL || force) {
    if (m_thread != NULL) {
      m_thread = NULL;
    }
#endif
    if (force && m_checker) {
      m_checker->SetCheckLatter("字典线程被强制重启");
    }

    pthread_create(&m_thread, NULL, DictThreadProc, (void*)this);
  }
#endif
  else {
    log_info("[DICT]字典线程已经启动,不要重复启动");
  }
  return true;
}

bool DictUpdateManager::CheckConnectionLost() {
  return kConnectLost == m_connect_status ||
         kConnectLostKicked == m_connect_status;
}
bool DictUpdateManager::CheckConnectionRecover() {
  if (kConnectRecover == m_connect_status) {
    m_connect_status = kConnectNormal;
    return true;
  }
  return false;
}
bool DictUpdateManager::NotifyFinished(bool all) {
  /*
          all:true,表示全部加载,false表示部分加载,只加载最近批次的数据

          将指定类型的字典数据从文件中加载到内存中
          外部需要实现DictLoaderListener来收取每条记录
          */

  SCOPED_LOCK(m_lock);
  bool has_update = all ? true : false;
  if (!has_update) {
    has_update = m_dict_manager.IsAnyDataUpdate();
  }
  if (!has_update) return false;

  BS_METRICS("[DICT]更新到内存容器");

  // m_loadorder保存了全部字典的加载顺序,此顺序与原先的DictLoader::LoadPart1顺序一致
  // 从映射文件中加载全部/部分记录,更新到对应的内存容器中
  int issue_count = m_dict_manager.SyncToMemory(all);

  if (all) {
    CBondContainer::instance().CalcIssueDateRange();
  }
  // 通知UI,all为true表示正在初次加载,此时无需通知
  if (!all) {
    if (IsDataUpdated(kDictTypeBondList) ||
        IsDataUpdated(kDictTypeBondPrimaryInfo) ||
        IsDataUpdated(kDictTypeBondAdditionInfo) ||
        IsDataUpdated(kDictTypeBondOptionInfo) ||
        IsDataUpdated(kDictTypeBondAuctionInfo) ||
        IsDataUpdated(kDictTypeBondCDCAuth)) {
      CBondContainer& cont = CBondContainer::instance();
      cont.CalcRemainingDate(ServerTimeMgr::instance().serverTime());
      cont.CalcBondExpireInHoliday();
      if (issue_count > 0) cont.CalcIssueDateRange();
      /*	CBusinessShowInfo iAuth;
                      AfxGetUserAccountManager().GetPrimaryMarketAuth_All(iAuth);
                      for (std::vector<CShowUnitInfo>::iterator it =
         iAuth.m_AllowList.begin(); it != iAuth.m_AllowList.end(); it++){
                              AfxGetHYQuotationMap(it->m_BrokerID.c_str()).UpdateBQMap();
                      }
                      log_debug("NotifyWndRcvInfo--eFT_PL_FILE_UPDATE");
          AfxGetBondReceiver().NotifyWndRcvInfo(eFT_PL_FILE_UPDATE,
         NULL);//通知UI 更新BondList*/
    }

    if (IsDataUpdated(kDictTypeMacroIndex)) {
      log_debug("NotifyWndRcvInfo--eFT_PL_MICRO_UPDATE");
      //	AfxGetBondReceiver().NotifyWndRcvInfo(eFT_PL_MICRO_UPDATE,
      //NULL);//通知UI 更新宏观数据
    }

    if (IsDataUpdated(kDictTypeFinancialCompany)) {
      log_debug("NotifyWndRcvInfo--eFT_PL_FINANCIALCOMP_UPDATE");
      //	AfxGetBondReceiver().NotifyWndRcvInfo(eFT_PL_FINANCIALCOMP_UPDATE,
      //NULL);//通知UI 更新机构列表
    }

    if (IsDataUpdated(kDictTypeBondSpecialLabel)) {
      log_debug("NotifyWndRcvInfo--eFT_PL_BOND_SPECIAL_LABEL_UPDATE");
      //	AfxGetBondReceiver().NotifyWndRcvInfo(eFT_PL_BOND_SPECIAL_LABEL_UPDATE,
      //NULL);//通知UI 更新特殊标签列表
    }
  }
  // 清理最近的一个更新包带来的记录印记
  m_dict_manager.ClearUpdateSet();
  m_check_results.clear();
  return true;
}
void DictUpdateManager::SetState(kState state) {
  /*
                            Exit
                                  ^
                                  |
          Req<====>Idle<====>Chk

  共有4种状态,
  Idle:当前啥也不做,空转,每间隔60s打印一条内部状态日志
  Req:当前正在处理请求,结束后,转换为Idle
  Chk:当前正在检查字典宝,结束后,转换为Idle
  Exit:当前正在结束
  */
  if (m_state == kStateReq) {
    BASSERT(state == kStateReq || state == kStateIdle);
  }
  if (m_state == kStateChk) {
    BASSERT(state == kStateChk || state == kStateIdle);
  }
  if (m_state != state) {
    log_info("[DICT]状态:[%s]  持续时间 %8d ms，切换到状态:[%s]",
             GetStateName(m_state), qb::base::GetHighResTime() - m_state_start,
             GetStateName(state));
  }
  m_state = state;
  m_state_start = qb::base::GetHighResTime();
}
void DictUpdateManager::ExportData(const char* fullDirectory) {
  SCOPED_LOCK(m_lock);
  bool unloaded = m_unloaded;
  if (unloaded) LoadDictCache(true);
  m_dict_manager.ExportData(fullDirectory);
  if (unloaded) LoadDictCache(false);
}
void DictUpdateManager::DumpStatus(int mode) {
  m_dict_manager.DumpStatus(mode);
}
bool DictUpdateManager::IsDataUpdated(kDictType type) {
  return m_dict_manager.IsDataUpdated(type);
}
void DictUpdateManager::BuildSpellIndex() {
  class SpellCollector {
    SpellSearch::Collector* m_collector;

   public:
    SpellCollector(bool build) : m_collector(NULL) {
      if (BOOLTEST(kProfileEnableBondFinder))
        m_collector = new SpellSearch::Collector(build);
    }
    ~SpellCollector() {
      if (m_collector) delete m_collector;
    }
    void AddNameChange(int id, const char* oldName, int oldlen,
                       const char* name, int len) {
      if (m_collector)
        m_collector->AddNameChange(id, oldName, oldlen, name, len);
    }
    void AddNameEnChange(int id, const char* oldName, int oldlen,
                         const char* name, int len) {
      if (m_collector)
        m_collector->AddNameEnChange(id, oldName, oldlen, name, len);
    }
    void AddCodeChange(int id, const char* oldCode, int oldlen,
                       const char* code, int len) {
      if (m_collector)
        m_collector->AddCodeChange(id, oldCode, oldlen, code, len);
    }
    void Commit() {
      if (m_collector) m_collector->Commit();
    }
  };
  class BondContainerLock {
   public:
    BondContainerLock() { CBondContainer::instance().Lock(); }
    ~BondContainerLock() { CBondContainer::instance().UnLock(); }
  };
  if (m_index_init) return;

  m_index_init = true;
  MEM_METRIC("BuildSpellIndex");
  {
    int nsize = CBondContainer::instance().GetSize();
    int batch = 1024;
    const char* steps[] = {"wait", "collect", "commit", "build"};
    TIME_METRIC_DECLARE();
    TIME_METRIC_SET_METRICS(steps);
    BVIEW("[SPRIT_INIT] start BuildSpellIndex");
    {
      SpellCollector collector(true);
      BS_METRICS("[SPRIT_INIT]BuildSpellIndex bonds:%d batch:%d", nsize, batch);
      for (int i = 0; i < nsize && !m_eixting;) {
        {
          BondContainerLock lock;
          TIME_METRIC_ADD_TIME(0);
          for (int k = 0; k < batch && i < nsize && !m_eixting; k++, i++) {
            const CBondInfo& bi = CBondContainer::instance().ElementAtR(i);
            if (!bi.IsValidBond()) continue;
            const char* name = bi.GetBondShortName();
            const std::string& name_en = bi.GetBondShortNameEn();
            const char* code = bi.GetCombBondCode();
            collector.AddNameChange(i, "", 0, name, name ? strlen(name) : 0);
            collector.AddNameEnChange(i, "", 0, name_en.c_str(),
                                      name_en.size());
            collector.AddCodeChange(i, "", 0, code, code ? strlen(code) : 0);
          }
        }
        TIME_METRIC_ADD_TIME(1);
        collector.Commit();
        TIME_METRIC_ADD_TIME(2);
      }
    }
    TIME_METRIC_ADD_TIME(3);
  }
}
void DictUpdateManager::CheckDict() {
  std::vector<int> sequences;
  int capcity = 64 * 1024;
  char* buf = new char[capcity];
  memset(buf, 0, capcity);

  std::vector<int> emptyseq;
  sequences.swap(emptyseq);
}
void DictUpdateManager::DictThreadRun() {
  /*
          字典线程,用于不断的处理解压后的更新包,并更新到对应的
          默认每次处理200ms时间的更新包请求
          当还没有完成的时候,是快速模式,等待1ms,当完成了则是慢速模式等待100ms
          */
  THREAD_NAME("Loop[DictUpdate]");
  //	log_info("[DICT]字典线程启动:%d",GetCurrentThreadId());
  DictThreadGuard::Get().OnStart();
  // ss::base::DeadLockCheck::AddThread("Loop[DictUpdate]");
  if (m_checker == NULL) m_checker = new DickChecker(m_dict_manager);

  if (m_needSyncFromServer) BuildSpellIndex();

  const int wait_fast = 1, wait_medium = 30, wait_slow = 100;
  DWORD start_tick = qb::base::GetHighResTime();
  DWORD last_tick = qb::base::GetHighResTime();
  while (!m_eixting) {
    int wait = wait_slow;
    DWORD now = qb::base::GetHighResTime();
    switch (GetState()) {
      case kStateReq:
        wait = wait_medium;
        if (ProcessRawPackets(false, now /*,200ms*/) > 0) wait = wait_fast;
        break;
      case kStateChk:
        OnCheck();
        break;
      case kStateIdle:
        if (m_signal.GetSignal() > 0)
          OnSignal(now);
        else if (CheckConnectionRecover())
          OnRecover(now);
        else
          OnIdle();
        break;
    }

    QThread::msleep(wait);
    if (wait == wait_slow && now > last_tick + 60000) {
      last_tick = now;
      log_info(
          "[DICT]字典线程已经运行%d秒 CheckQueue:%d RawQueue:%d State:%s "
          "Req:%d",
          (last_tick - start_tick) / 1000, m_check_queue.Size(),
          m_raw_queue->Size(), GetStateName(GetState()), m_req_count);
    }
    DictThreadGuard::Get().OnHeartbeat(now);
  }

  if (m_checker) delete m_checker;

  m_checker = NULL;
}
void DictUpdateManager::OnSignal(DWORD tick) {
  if (tick > m_signal.GetTick() + m_signal_gap) {
    m_signal_gap = qb::base::Rand::RandInt(10000, 150000);
    SendFirstRequest(kPullTypeOnSignal);
    LoadDictCache(true);
  }
}
void DictUpdateManager::OnRecover(DWORD tick) {
  if (m_last_conreq_time <= 0 || (tick >= m_last_conreq_time + m_conreq_gap)) {
    m_last_conreq_time = tick;
    SendFirstRequest(kPullTypeOnConnectRecover);
    LoadDictCache(true);
  }
}
void DictUpdateManager::OnCheckFinished() {
  SetState(kStateIdle);
  m_check_results.clear();
  m_checker->GetCheckResult(m_check_results);
  if (BOOLTEST(kProfileDictRequestOnFailure) &&
      (m_check_results.size() > 0 || m_dict_manager.HasEmptyDict())) {
    for (auto it = m_check_results.begin(); it != m_check_results.end(); ++it) {
      DictCheckInfo& info = it->second;
      const DictDesc& dd = DictDesc::GetStockDictDesc(it->first);
      log_error("[DICT][CHECK]字典:%16s 检查失败,重新从包:%s-%d 开始拉取 %s",
                dd.GetVersion().c_str(), info.ack.m_Md5, info.ack.m_UpdateId,
                info.ack.m_UpdateId <= 0 ? "字典重建" : "");

      char szMd5[64] = {0};
      if (info.ack.m_UpdateId > 0 &&
          BOOLTEST(kProfileDictCheckUpdateIdHistory)) {
        int idHist = m_dict_manager.GetUpdateIdBefore(
            it->first, info.ack.m_UpdateId, szMd5, _countof(szMd5));
        if (idHist > 0) {
          log_error("[DICT][CHECK]字典:%16s 调整UpdateId:%6d ==> %6d",
                    dd.GetVersion().c_str(), info.ack.m_UpdateId, idHist);
          info.ack.m_UpdateId = idHist;
          FIELDCOPY(info.ack.m_Md5, szMd5);
        } else {
          info.ack.m_UpdateId = 0;
          FIELDCOPY(info.ack.m_Md5, "");
          log_error("[DICT][CHECK]字典:%16s 找不到更新包历史,准备全量更新",
                    dd.GetVersion().c_str());
        }
      }
      if (info.ack.m_UpdateId <= 0 && BOOLTEST(kProfileDictRecreateOnError)) {
        m_dict_manager.RecreateTable(dd.GetType());
      }
    }
    SendFirstRequest(kPullTypeOnCheckFailure);
  } else {
    bool server_error = m_checker->HasServerError();
    bool sync_error = m_check_results.size() > 0;
    if (!server_error && !sync_error) {
      log_warning("[DICT][CHECK]字典检查全部完成,与服务器状态一致");
    } else {
      log_error("[DICT][CHECK]字典检查全部完成,后台%s数据错误 前端同步错误:%d",
                server_error ? "有" : "无", m_check_results.size());
    }
  }
  m_checker->PrintDictState(false);
}
void DictUpdateManager::OnCheck() {
  /*
          检查的响应消息可能丢失，收不到，导致无法从kStateChk恢复到kStateIdle
  */
  DictCheckAck* ack = (DictCheckAck*)m_check_queue.Pick();
  if (ack) {
    if (m_checker->CheckFinished(ack->reqid, ack->ack)) {
      OnCheckFinished();
    }
    delete ack;
  } else if (qb::base::GetHighResTime() > m_state_start + m_max_check_time) {
    // 字典检查超时后,会在一定时间内立即启动检查
    DWORD now = qb::base::GetHighResTime();
    if (now > m_check_latter + 60000) {
      char buf[256] = {0};
      const char* fmt = "上次字典请求超时,最大:%8d ms 实际:%8d ms,";
      FMTBUF(buf, fmt, m_max_check_time,
             qb::base::GetHighResTime() - m_state_start);
      m_checker->SetCheckLatter(buf);
      m_check_latter = now;
    }
  }
}
void DictUpdateManager::OnIdle() {
  if (GetState() == kStateIdle) {
    if (NotifyFinished(false))
      LoadDictCache(false);  // 请求结束后,清空内存,文件保存好

    // 开始字典检测,只有当满足时间条件的时候才会真正开始
    // 当字典更新时有至少1个字典的更新响应包出现超时,会触发紧急检测
    if (m_req_progress->IsNeedEmergentCheck()) {
      m_req_progress->SetEmergentCheck(false);

      char buf[256] = {0};
      const char* fmt = "字典请求中有 %2d 个超时 ";
      FMTBUF(buf, fmt, m_req_progress->GetSkipCount());
      m_checker->SetCheckLatter(buf);
    }
    if (m_checker->StartCheck()) {
      SetState(m_checker->IsChecking() ? kStateChk : kStateIdle);
    }
  }
}
void DictUpdateManager::LoadDictCache(bool load) {
  /*
  加载所有字典,会重建cache,和创建映射文件
  */
  qb::base::HighResTime time;
  SCOPED_LOCK(m_lock);
  if (!load) m_dict_manager.Clear();  // 请求结束后,清空内存,文件保存好
  m_unloaded = !load;
  log_info("[DICT]%s last %d", load ? "ClearDicts" : "LoadDicts",
           time.elapsed().microsec());
}
int DictUpdateManager::ProcessRawPackets(bool all, DWORD nowTime,
                                         DWORD batchTime, DWORD* pWorkTime) {
  /*
          统计表明:平均每个消息包处理需要消耗1735us时间(实际可能更短)
          旧策略:每个消息处理完毕后,都要检测是否超出批处理时间
          新策略:每间隔10个消息后才检测,减少GetTickCount()的开销
          旧策略中:GetTickCount的开销占据ProcessRawPackets的10%
          批处理时间不要求非常精准
  */
  int processed = 0;
  int checked = 0;
  const int max_check = 10;
  DWORD start = nowTime;
  DWORD now = start;
  DWORD end = now + batchTime;
  DWORD max_req = m_max_rsp_time;
  bool simulate =
      false;  // 用于模拟超时的消息中前面5个消息不当做超时,之后的才当作超时
  while (1) {
    // 如下代码中不能出现continue,否则末尾的不能被跳过的代码被跳过导致逻辑出错
    UpdatePacket* raw = (UpdatePacket*)m_raw_queue->Pick(20);
    if (raw == NULL) {
      if (GetState() == kStateReq && now > m_req_time + max_req) {
        bool neednext = true;
        if (simulate && m_current_type == kDictTypeBondList &&
            m_req_progress->GetMsgCount(m_current_type) < 5)
          neednext = false;
        if (neednext) {
          // 当前类型的请求超时,
          m_req_progress->AddSkipDict(m_current_type);
          log_info("[DICT]字典请求超时,%s,超时:%8d ms 已收到:%d",
                   DictDesc::GetStockDictDesc(m_current_type).GetName().c_str(),
                   now - m_req_time, m_current_msgcount);
          SendNextRequest(m_current_type);
        }
      }
      break;
    }

    // BASSERTS(m_unloaded == false, "字典必须没有被卸载");
    bool skipped = raw->type < m_current_type ? true : false;
    if (!skipped && simulate && raw->type == kDictTypeBondList) {
      if (m_req_progress->GetMsgCount(raw->type) >= 5) {
        skipped = true;
      }
    }
    if (skipped) {
      if (raw->fileBuf)
        m_dict_manager.UpdatePack(raw, all);
      else {
        log_error("[DICT]延迟的字典到达响应,文件内容为空!!!!!!");
      }
      log_error(
          "[DICT]延迟的字典到达响应,更新:延迟:%6d ms PatchName:%s szMd5:%s "
          "updateId:%8d,当前:%2d-%16s,期望:%2d-%16s",
          m_req_progress->GetRspTime(raw->type), raw->szPatchName, raw->szMd5,
          raw->updateId, raw->type,
          DictDesc::GetStockDictDesc(raw->type).GetName().c_str(),
          m_current_type,
          DictDesc::GetStockDictDesc(raw->type).GetName().c_str());
    } else {
      if (raw->finish) {
        SendNextRequest(raw->type);
      } else {
        SCOPED_LOCK(m_lock);
        m_current_msgcount++;
        m_req_progress->SetMsgCount(raw->type, m_current_msgcount);
        if (raw->fileBuf)
          m_dict_manager.UpdatePack(raw, all);
        else {
          BASSERT(raw->fileBuf);
        }
        if (all) m_progresser.AddData(raw->origsize);
      }
    }
    if (m_history) {
      m_history->Add(raw);
    }
    m_allocator.Release(raw);
    processed++;
    checked++;
    if (checked >= max_check) {
      checked = 0;
      now = qb::base::GetHighResTime();
      if (now >= end) break;
    }
  }

  if (processed > 0 && pWorkTime) {
    *pWorkTime = *pWorkTime + qb::base::GetHighResTime() - start;
  }
  return processed;
}
void DictUpdateManager::SendNextRequest(kDictType type) {
  BASSERT(type >= kDictTypeBondList && type < kDictTypeCount);
  BASSERT(GetState() == kStateReq);
  kDictType next_type = (kDictType)(type + 1);
  if (next_type != kDictTypeCount) {
    SendDictRequest(next_type);
  } else {
    log_debug("[DICT]REQUEST FINISH,packet_count:%d", m_raw_queue->Size());
    SCOPED_LOCK(m_lock);
    SetState(kStateIdle);
  }
}
void DictUpdateManager::WaitForFinish() {
  /*
  程序退出的时候调用,用于退出时处理数据和结束线程
  */
  m_eixting = true;
#ifdef WIN32
  if (m_thread) {
    WaitForSingleObject(m_thread, 10000);
    DictThreadGuard::Get().StopCheck();
    CloseHandle(m_thread);
    m_thread = NULL;
  }
#else
#ifdef CCORE_LINUX
  if (m_thread != -1) {
    pthread_cancel(m_thread);
    pthread_join(m_thread, NULL);
    DictThreadGuard::Get().StopCheck();
    m_thread = -1;
  }
#else
  if (m_thread != NULL) {
    pthread_cancel(m_thread);
    pthread_join(m_thread, NULL);
    DictThreadGuard::Get().StopCheck();
    m_thread = NULL;
  }
#endif
#endif
}

bool DictUpdateManager::checkLoadStatus() {
  if (m_event->Wait(200000) == kEventSignaled && !m_needSyncFromServer)
    return true;
  return false;
}

DictUpdateManager& DictUpdateManager::instance() {
  static DictUpdateManager gs_dict_updater;
  return gs_dict_updater;
}

STDMETHODIMP_(void)
DictUpdateManager::OnRecvNetStatus(qb::NetStatus status, short retcode) {
  if (eNetStatus_Done == status && 0 == retcode) {
  }
}

STDMETHODIMP_(void)
DictUpdateManager::OnResponse(THIS_ int32_t msgtype, qb::ProcessMessage msp) {
  const char* pData = msp.getBuff();
  QB_CHECK_RETURN_VOID1(pData);

  if (msgtype == E_FID_CT_CLIENT_FILEUPDATE_MSG_REQ) {
    xMsgHeadAck_t xHead;
    memcpy(&xHead, pData, sizeof(xHead));
    xHead.msgtype = ntohl(xHead.msgtype);
    xHead.retcode = ntohs(xHead.retcode);
    xHead.idx = ntohs(xHead.idx);
    xHead.num = ntohs(xHead.num);

    if (xHead.retcode == E_RC_OK) {
      ProcessFileUpdate_DATA(pData + sizeof(xHead), xHead);
    } else {  // E_RC_NODATA无数据，不用更新
      xRespFileUpdateMsgMass_t xBody;
      memset(&xBody, 0, sizeof(xBody));
      memcpy(&xBody, pData + sizeof(xHead), sizeof(xBody));
      DictUpdateManager::instance().OnPackReceived(msgtype, &xBody);
    }
  } else if (msgtype == E_FID_CT_CLIENT_FILEUPDATE_FIN) {
    xMsgHeadAck_t xHead;
    memcpy(&xHead, pData, sizeof(xHead));
    xHead.msgtype = ntohl(xHead.msgtype);
    xHead.retcode = ntohs(xHead.retcode);
    xHead.idx = ntohs(xHead.idx);
    xHead.num = ntohs(xHead.num);

    xRespFileUpdateMsgMass_t xBody;
    memset(&xBody, 0, sizeof(xBody));
    memcpy(&xBody, pData + sizeof(xHead), sizeof(xBody));
    DictUpdateManager::instance().OnPackReceived(msgtype, &xBody);
  }
}

STDMETHODIMP_(void) DictUpdateManager::OnForceQuit(THIS_ int32_t connect_id) {
  OnConnectSignal(eFT_PL_FORCE_QUIT);
}

STDMETHODIMP_(void) DictUpdateManager::OnDisConnect(THIS_ int32_t connect_id) {
  OnConnectSignal(eFT_PL_HEART_BREAK_SYS_NOK);
}

STDMETHODIMP_(void) DictUpdateManager::OnRecover(THIS_ int32_t connect_id) {
  OnConnectSignal(eFT_PL_HEART_BREAK_SYS_OK);
}

void DictUpdateManager::ProcessFileUpdate_DATA(const char* RecBuf,
                                               xMsgHeadAck_t xHead) {
  xRespFileUpdateMsgMass_t recData;
  FormatFileUpdate_DATA(RecBuf, xHead.msgtype, recData);
  if (recData.fileLen > 0) {
    static xRespFileUpdateMsgMass_t glrecData;
    if (xHead.idx == 0 && xHead.num > 0)  // 第一个包
    {
      memset(&glrecData, 0x00, sizeof(xRespFileUpdateMsgMass_t));
      memcpy(&glrecData, &recData,
             sizeof(xRespFileUpdateMsgMass_t) - sizeof(char*));
      int nSize = recData.fileLen * xHead.num;
      glrecData.fileBuf =
          new (std::nothrow) char[nSize];  // 先分配好总的内存空间
      if (glrecData.fileBuf) {
        memset(glrecData.fileBuf, 0, nSize * sizeof(char));
        memcpy(glrecData.fileBuf, recData.fileBuf, recData.fileLen);
      }
    } else {  // 中间的包
      memcpy(glrecData.fileBuf + glrecData.fileLen, recData.fileBuf,
             recData.fileLen);
      glrecData.fileLen += recData.fileLen;
    }

    if (xHead.num == xHead.idx + 1)  // 最后一个包
    {
      DictUpdateManager::instance().OnPackReceived(
          E_FID_CT_CLIENT_FILEUPDATE_MSG_REQ, &glrecData);

      if (glrecData.fileBuf) delete[] (char*)glrecData.fileBuf;
      memset(&glrecData, 0, sizeof(xRespFileUpdateMsgMass_t));
    }
  }
}

void DictUpdateManager::FormatFileUpdate_DATA(
    const char* RecBuf, int msgtype, xRespFileUpdateMsgMass_t& RecData) {
  int LenIdx = 0;
  memset(&RecData, 0x00, sizeof(RecData));

  memcpy(&RecData.sequence, RecBuf, sizeof(int));  //
  LenIdx += sizeof(int);
  RecData.sequence = ntohl(RecData.sequence);
  memcpy(&RecData.updateId, RecBuf + LenIdx, sizeof(int));  //
  LenIdx += sizeof(int);
  RecData.updateId = ntohl(RecData.updateId);
  memcpy(&RecData.szCategory, RecBuf + LenIdx, sizeof(RecData.szCategory));  //
  LenIdx += sizeof(RecData.szCategory);
  memcpy(&RecData.szPatchName, RecBuf + LenIdx,
         sizeof(RecData.szPatchName));  //
  LenIdx += sizeof(RecData.szPatchName);
  memcpy(&RecData.szFileName, RecBuf + LenIdx, sizeof(RecData.szFileName));  //
  LenIdx += sizeof(RecData.szFileName);
  memcpy(&RecData.origsize, RecBuf + LenIdx, sizeof(RecData.origsize));  //
  LenIdx += sizeof(RecData.origsize);
  memcpy(&RecData.szMd5, RecBuf + LenIdx, sizeof(RecData.szMd5));  //
  LenIdx += sizeof(RecData.szMd5);
  memcpy(&RecData.fileLen, RecBuf + LenIdx, sizeof(RecData.fileLen));  //
  LenIdx += sizeof(RecData.fileLen);
  if (RecData.fileLen > 0) {
    RecData.fileBuf = (char*)RecBuf + LenIdx;
  }
}

//////////////////////////////////////////////////////////////////////////
ProgressNotifier::ProgressNotifier()
    : m_start(0),
      m_end(0),
      m_processed_size(0),
      m_total_size(0),
      m_real_total_size(0),
      m_progress(0),
      m_recv_finished(false),
      m_stop_ratio(0.3) {}

void ProgressNotifier::Init(int start, int end) {
  m_start = start;
  m_end = end;
  m_progress = start;
  m_total_size = 0;
  m_real_total_size = 0;
  m_recv_finished = false;
}

void ProgressNotifier::SetDataSize(int bytes) { m_total_size = bytes; }

void ProgressNotifier::AddTotal(int bytes) {
  if (m_recv_finished) return;
  if (bytes >= 0)
    m_real_total_size += bytes;
  else
    m_recv_finished = true;
}

void ProgressNotifier::AddData(int bytes) {
  /*
          分为两个阶段
          第一阶段,m_recv_finished为false,表示数据收取还在继续,进度是模糊的
          30%
          第二阶段,m_recv_finished为true,表示数据收取完毕,进度是精确的
          */
  if (m_total_size <= 0) return;
  m_processed_size += bytes;
  double ratio = 0;
  if (m_recv_finished && m_real_total_size > 0)
    ratio = m_processed_size * 1.0 / m_real_total_size;
  else
    ratio = m_processed_size * m_stop_ratio / m_total_size;
  int progress = m_start + (m_end - m_start) * ratio;
  if (progress > m_progress) {
    m_progress = progress;
    m_progress = m_progress >= m_end ? (m_end - 1) : m_progress;
    // PostMessage(m_window, m_message, m_progress, 0);
  }
}

void ProgressNotifier::Finish() {
  // PostMessage(m_window, m_message, m_end, 0);
}
